package com.hp.bon.sgw.core.channel;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.dom4j.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hp.bon.security.BonEncrypt;
import com.hp.bon.security.BonSignature;
import com.hp.bon.sgw.bean.ProtocalConvertResult;
import com.hp.bon.sgw.bean.SysParamBean;
import com.hp.bon.sgw.bean.XmlMessage;
import com.hp.bon.sgw.convert.DataExchangerImpl;
import com.hp.bon.sgw.core.BuzProcessorManager;
import com.hp.bon.sgw.core.CallParam;
import com.hp.bon.sgw.core.InprogressMessageQueue;
import com.hp.bon.sgw.core.channel.handler.DCCAsynCallHandler;
import com.hp.bon.sgw.core.processor.Processor;
import com.hp.bon.sgw.domain.MessageDetail;
import com.hp.bon.sgw.domain.Node;
import com.hp.bon.sgw.domain.SGWChannel;
import com.hp.bon.sgw.service.BONNetCtrlService;
import com.hp.bon.sgw.service.ConfigListener;
import com.hp.bon.sgw.service.ConfigService;
import com.hp.bon.sgw.service.FluxControlService;
import com.hp.bon.sgw.start.SpringContext;
import com.hp.bon.sgw.threads.BusinessRecordCacheManager;
import com.hp.bon.sgw.threads.MessageDetailCacheManager;
import com.hp.bon.sgw.util.Comutils;
import com.hp.bon.sgw.util.dcc.DCCMsgDecorator;
import com.hp.bon.sgw.util.dcc.DccAvpHelper;
import com.hp.bon.sgw.wh.service.UsageLimitConfigService;
import com.hp.ocs.diameter.DiameterUtils;
import com.hp.ocs.diameter.message.AVP;
import com.hp.ocs.diameter.message.AVP_FailedAVP;
import com.hp.ocs.diameter.message.AVP_OctetString;
import com.hp.ocs.diameter.message.AVP_UTF8String;
import com.hp.ocs.diameter.message.AVP_Unsigned32;
import com.hp.ocs.diameter.message.Message;
import com.hp.ocs.diameter.message.MessageDictionary;
import com.hp.ocs.diameter.node.Capability;
import com.hp.ocs.diameter.node.Connection;
import com.hp.ocs.diameter.node.InvalidSettingException;
import com.hp.ocs.diameter.node.MessageDispatcher;
import com.hp.ocs.diameter.node.NodeManager;
import com.hp.ocs.diameter.node.NodeSettings;
import com.hp.ocs.diameter.node.NodeStatUtils;
import com.hp.ocs.diameter.node.NotARequestException;
import com.hp.ocs.diameter.node.NotAnAnswerException;
import com.hp.ocs.diameter.node.Peer;
import com.hp.ocs.diameter.node.PeerFactory;
import com.hp.ocs.diameter.node.StaleConnectionException;
import com.hp.ocs.diameter.node.UnsupportedURIException;

public class DCCChannel extends NodeManager implements IChannel, ConfigListener {
	public static final Logger LOG = LoggerFactory.getLogger(DCCChannel.class);
	private SGWChannel sgwChannel;
	private ConfigService confSrv;
	private Map<String, Node> persistNodes = new ConcurrentHashMap<String, Node>();
	private Map<String, Node> allowedNodes = new ConcurrentHashMap<String, Node>();
	public volatile int dwrTimeOut;
	public volatile int dwrTimeOutCount;
	public volatile int reConnectTime;
	protected volatile int ccrSendTimeout;
	protected volatile int ccrRepeatCount;
	public volatile boolean checkStrickCheckCCR = false;
	public volatile boolean printMessageDetail = true;

	private final Set<String> monitorConfKeys = new HashSet<String>();

	public DCCChannel(SGWChannel sgwChannel) {
		this.sgwChannel = sgwChannel;
		this.confSrv = SpringContext.getInstance().getConfigService();
		this.confSrv.addConfigListener(this);
	}

	private void init() {
		NodeSettings settings = buildDCCNodeParams(this.sgwChannel, this.confSrv);
		initSettings(settings);
		setEnableFailedAVP(true);
		setCheckMessageFormat(this.checkStrickCheckCCR);
		int taskQueueSize = this.confSrv.getIntValue("Diameter.TaskQueueSize", 1000);
		setRetransmitParameter(this.ccrRepeatCount, this.ccrSendTimeout * 1000);

		setReconnectRetryTime(this.reConnectTime);
		String tcpSendBuffer = this.confSrv.getValue("Diameter.TCP.SendBuffer", "2M");
		String tcpRecvBuffer = this.confSrv.getValue("Diameter.TCP.ReceiveBuffer", "2M");
		Integer senderQueueSize = Integer.valueOf(this.confSrv.getIntValue("Diameter.TCP.SenderQueueSize", 10000));

		int minThread = this.confSrv.getIntValue("business.min_process", 20);
		int maxThread = this.confSrv.getIntValue("business.max_process", 50);

		useThreadPool(minThread, maxThread, 300, taskQueueSize);
		setConnectionTimeout(this.dwrTimeOutCount, this.dwrTimeOut, 0L);
		setTcpBuffer(DiameterUtils.getCapacityBytes(tcpSendBuffer, -1), DiameterUtils.getCapacityBytes(tcpRecvBuffer, -1));
		setSenderQueueSize(senderQueueSize.intValue());
	}

	public void start() throws IOException {
		LOG.info("start dcc channel:" + this.sgwChannel.toString() + " ...");
		init();

		initParams();
		loadConfParams();

		super.start();
		this.sgwChannel.setStatus(1);
		this.confSrv.updateChannelStatus2DB(this.sgwChannel.getId(), 1);

		loadPeerNodes();
	}

	public void stop() {
		LOG.info("stop dcc channel:" + this.sgwChannel.toString() + " ...");
		NodeStatUtils.getInstance().stop();
		super.stop();
		this.sgwChannel.setStatus(0);
		this.confSrv.updateChannelStatus2DB(this.sgwChannel.getId(), 0);
	}

	public int onConnectionCapbilityExchanging(Peer peer) {
		int result = super.onConnectionCapbilityExchanging(peer);
		if (result != 2001)
			return result;
		if (!Peer.isValidHostID(peer.hostID())) {
			LOG.warn("3010,invalid hostId: " + peer.hostID());
			return 3010;
		}

		peer.capabilities().addAcctApp(4);

		result = validateOnExchangeCapability(peer.hostID(), peer.host());
		if (result != 2001)
			return result;
		return 2001;
	}

	public void onConnectionReady(Peer peer) {
		LOG.info("connection ready ,peer " + peer.hostID());
		peer.setAttachment("OK");

		Node node = this.confSrv.getNodeByName(peer.hostID());
		if (node != null) {
			updateAndInfoNodeStatus(node, true);
			if (SpringContext.getInstance().getPertriBean().getIfOnNet())
				if (node.getNodeType() == 1)
					BONNetCtrlService.getInstance().sendActive();
				else
					BONNetCtrlService.getInstance().sendNodeActiveOrDeActive(node, true);
		}
	}

	public void onConnectionClosed(Peer peer) {
		super.onConnectionClosed(peer);

		if ((peer == null) || (!"OK".equals(peer.getAttachment()))) {
			LOG.warn("can't connect peer ,errcode 9 , node is " + peer.hostID());
			return;
		}
		peer.setAttachment(null);

		Node node = this.confSrv.getNodeByName(peer.hostID());
		if (node != null) {
			if (SpringContext.getInstance().getPertriBean().getIfOnNet()) {
				BONNetCtrlService.getInstance().sendNodeActiveOrDeActive(node, false);
			}
			updateAndInfoNodeStatus(node, false);
		}
	}

	public int validateOnExchangeCapability(String hostID, String hostIP) {
		Node node = this.confSrv.getNodeByName(hostID);

		if (node == null) {
			LOG.warn("3010|hostId not exist:" + hostID);
			return 3010;
		}

		if (node.getChannelId() != this.sgwChannel.getId()) {
			LOG.warn("3010|hostId not allowed:" + hostID);
			return 3010;
		}

		if (isCongestionToReject()) {
			return 5003;
		}

//		if (isFluxContrlToReject(hostID)) {
//			return 4001;
//		}

		if (!node.checkAccessedIP(hostIP)) {
			LOG.warn("AuthFailed:NE IP not matched,expected is " + node.getIP() + ",but now is " + hostIP);
			return 3010;
		}
		return 2001;
	}

	public boolean notifySendFailed(Message arg0, Peer peer, MessageDispatcher.SendFailedReason res) {
		String sessionID = new AVP_UTF8String(arg0.find(263)).queryValue();
		LOG.warn("SendFailed:" + peer.hostID() + "|" + sessionID + "|" + arg0.toString() + "|" + res.toString());
		return super.notifySendFailed(arg0, peer, res);
	}

	protected void handleRequest(Message request, Peer peer) {
		double persent = getSGWChannel().getRecvQueuePercent();
		double MIN = SpringContext.getInstance().getConfigService().getIntValue("business.fluxctrl.receive_queue_per", 50) * 1.0D / 100.0D;
		if ((persent > MIN) && (Math.random() < (persent - MIN) / (1.0D - MIN))) {
			directReplyCCA1(request, peer, UsageLimitConfigService.getIntValue("DIAMETER_RESULT_TOO_BUSY", 501), null);
			return;
		}

		AVP sessionIDAVP = request.find(263);
		if (sessionIDAVP == null) {
			AVP avp = new AVP_UTF8String(263, "SESSION_ID is null");
			AVP failedAVP = new AVP_FailedAVP(avp);
			directReplyCCA(request, peer, 5005, failedAVP);
			LOG.warn("missing Session-Id AVP,hop=" + request.header().hopByHopID());
			return;
		}
		String sessionID = AVP_UTF8String.queryValue(sessionIDAVP);
		if (sessionID.equals("")) {
			AVP avp = new AVP_UTF8String(263, "SESSION_ID is Blank");

			AVP failedAVP = new AVP_FailedAVP(avp);
			directReplyCCA(request, peer, 5002, failedAVP);
			LOG.warn("Session-Id is unknown,hop=" + request.header().hopByHopID());
			return;
		}
		if (this.confSrv.getBooleanValue(SysParamBean.KEY_AGREEMENT_DCC_CheckStrick, false)) {
			MessageDictionary md = request.getDictionary();
			int checkResult = md.checkFormat();
			if (checkResult != 2001) {
				AVP failedAVP = md.getFailedAVP();
				if (failedAVP == null) {
					failedAVP = request.getFailedAVP();
				}
				directReplyCCA(request, peer, checkResult, failedAVP);
				LOG.warn(sessionID + "|strick check CCR failed");
				return;
			}
		}
		if (!request.header().isProxiable()) {
			AVP failedAVP = request.getDictionary().getFailedAVP();
			directReplyCCA(request, peer, 3008, failedAVP);
			LOG.warn(sessionID + ",header().isProxiable() is false");
			return;
		}
		if (request.find(268) != null) {
			AVP failedAVP = request.getDictionary().getFailedAVP();
			directReplyCCA(request, peer, 5008, failedAVP);
			LOG.warn(sessionID + ",Result-Code AVP is not allowed in CCR");
			return;
		}
		AVP orighHostAVP = request.find(264);
		if (orighHostAVP == null) {
			AVP avp = new AVP_UTF8String(264, "No ORIGIN_HOST");
			AVP failedAVP = new AVP_FailedAVP(avp);
			directReplyCCA(request, peer, 5005, failedAVP);
			LOG.warn(sessionID + ",missing Orgin-Host AVP");
			return;
		}
		String orinHost = AVP_UTF8String.queryValue(orighHostAVP);
		String serviceContextId = null;
		AVP srvCtxAVP = request.find(461);
		int commdCode = request.header().commandCode();
		if (srvCtxAVP == null) {
			switch (commdCode) {
			case 258:
				serviceContextId = "SessionCtr.OCS@" + DccAvpHelper.getRealmPart(orinHost);
				break;
			case 274:
				serviceContextId = "SessionCtr.OCS@" + DccAvpHelper.getRealmPart(orinHost);
				break;
			default:
				AVP avp = new AVP_UTF8String(461, "No SERVICE_CONTEXT_ID");
				AVP failedAVP = new AVP_FailedAVP(avp);
				directReplyCCA(request, peer, 5005, failedAVP);
				LOG.warn("11104," + sessionID + ",missing Service_Context_Id AVP");
			}
		}
		if (serviceContextId == null) {
			serviceContextId = AVP_UTF8String.queryValue(srvCtxAVP);
		}

		StringBuilder errMsg = new StringBuilder();
		DCCAsynCallHandler asynHandler = new DCCAsynCallHandler(request, peer, this);
		CallParam callParam = new CallParam(sessionID, serviceContextId, request, peer.hostID(), asynHandler);

		boolean securitySucc = true;
		if (this.confSrv.getNodeByName(peer.hostID()).getNodeType() == 1) {
			int signResult = 0;
			int decResult = 0;
			switch (commdCode) {
			case 274:
				signResult = BonSignature.checkSignMessageNoSCID(request, orinHost);
				break;
			case 272:
				signResult = BonSignature.checkSignMessage(request, serviceContextId, orinHost);
				if (signResult == 0) {
					decResult = BonEncrypt.decrypteMessage(request, serviceContextId, orinHost);
				}
				break;
			case 273:
			}
			if (signResult != 0) {
				errMsg.append(signResult + "," + sessionID + ",check sign msg(request) failed,error code:" + signResult);
				LOG.warn(errMsg.toString());
				Message answer = directReplyCCAWithServiceResultCode(request, peer, signResult, null);
				securitySucc = false;

				callParam.setRetCode(signResult);
				callParam.setResultMsg(errMsg.toString());
				callParam.setResponse(answer);
				callParam.setToNode(callParam.getFromNode());
			}
			if (decResult != 0) {
				errMsg.append(signResult + "," + sessionID + ",decrypte msg(request) failed,error code:" + signResult);
				LOG.warn(errMsg.toString());
				Message answer = directReplyCCAWithServiceResultCode(request, peer, 10106, null);
				securitySucc = false;

				callParam.setRetCode(decResult);
				callParam.setResultMsg(errMsg.toString());
				callParam.setResponse(answer);
				callParam.setToNode(callParam.getFromNode());
			}
		}

		MessageDetail md = new MessageDetail();
		md.setBusinessCallId(callParam.getBussinessCallId());
		md.setSessionId(callParam.getSessionId());
		md.setServiceContextId(callParam.getServiceContextId());
		md.setMessageType("CCR");
		md.setOpType("recv");
		md.setHostId(peer.hostID());
		md.setMsgTime(new Date(request._rt0));
		md.setDetails(request.getDictionary().toString());
		MessageDetailCacheManager.getInstance().add(md);

		if (!securitySucc) {
			BusinessRecordCacheManager.getInstance().add(callParam);
		} else {
			int ckeckResult = BuzProcessorManager.getInstance().checkCallParamForDcc(callParam, errMsg);
			if (ckeckResult != 0) {
				LOG.warn(errMsg.toString());
				Message answer;
				if (ckeckResult > 10000) {
					answer = directReplyCCAWithServiceResultCode(request, peer, ckeckResult, null);
				} else {
					answer = directReplyCCA(request, peer, ckeckResult, null);
				}

				callParam.setRetCode(ckeckResult);
				callParam.setResultMsg(errMsg.toString());
				callParam.setResponse(answer);
				callParam.setToNode(callParam.getFromNode());
				BusinessRecordCacheManager.getInstance().add(callParam);
				return;
			}
		}

		BuzProcessorManager.getInstance().processRequest(callParam);
	}

	protected void handleAnswer(Message answer, Peer peer, Message req) {
		String sessionId = AVP_UTF8String.queryValue(answer.find(263));
		CallParam callParam = InprogressMessageQueue.getInstance().removeProgress(sessionId);

		StringBuilder errMsg = new StringBuilder();
		if (callParam == null) {
			LOG.warn("not find callParam for " + sessionId);
			return;
		}
		boolean securitySucc = true;

		int commdCode = answer.header().commandCode();
		switch (commdCode) {
		case 258:
			break;
		case 274:
			break;
		case 272:
			if (this.confSrv.getNodeByName(peer.hostID()).getNodeType() == 1) {
				int signResult = BonSignature.checkSignMessage(answer, callParam.getServiceContextId(), DccAvpHelper.getMessageOriginalHost(answer));
				if (signResult != 0) {
					errMsg.append(sessionId + ",check sign msg(answer) failed,error code:" + signResult);
					LOG.warn(errMsg.toString());

					callParam.setRetCode(signResult);
					callParam.setResultMsg(errMsg.toString());

					securitySucc = false;
				}

				int decResult = BonEncrypt.decrypteMessage(answer, callParam.getServiceContextId(), DccAvpHelper.getMessageOriginalHost(answer));
				if (decResult != 0) {
					errMsg.append(sessionId + ",decrypte sign msg(answer) failed,error code:" + decResult);
					LOG.warn(errMsg.toString());

					callParam.setRetCode(decResult);
					callParam.setResultMsg(errMsg.toString());

					securitySucc = false;
				}
			}
			break;
		default:
			LOG.warn(answer.toString() + " UnExpected CommdCode:" + commdCode);
		}

		if (securitySucc) {
			callParam.setRecvResponse(answer);
			int s_rc = DccAvpHelper.getCCAMessageServiceResultCode(answer);
			callParam.setServiceResultCode(s_rc);
		}

		Processor p = callParam.getProcessor();
		p.endProcess(callParam);

		MessageDetail md = new MessageDetail();
		md.setBusinessCallId(callParam.getBussinessCallId());
		md.setSessionId(callParam.getSessionId());
		md.setServiceContextId(callParam.getServiceContextId());
		md.setMessageType("CCA");
		md.setOpType("recv");
		md.setHostId(peer.hostID());
		md.setMsgTime(new Date(answer._rt0));
		md.setDetails(answer.getDictionary().toString());
		MessageDetailCacheManager.getInstance().add(md);
	}

	public int sendRequest(CallParam callParam) {
		Node recvNode = callParam.getToNode();
		Peer toPeer = getNodeConnect(recvNode.getHostId());
		Message newMessage;
		if ((callParam.getRequest() instanceof Message)) {
			Message req = (Message) callParam.getRequest();
			newMessage = new Message();

			DccAvpHelper.copyMessageHeader(req.header(), newMessage.header());
			DccAvpHelper.copyMessageAVP(req, newMessage);
		} else if ((callParam.getRequest() instanceof XmlMessage)) {
			newMessage = genBonCallCCR(callParam);
			if (newMessage == null)
				return -1;
		} else {
			LOG.warn("not expected Message type:" + callParam.getRequest().getClass().getSimpleName());
			return -1;
		}
		if (newMessage.header().commandCode() == 272) {
			DCCMsgDecorator.getInstance().decorateMsg(newMessage, callParam.getFromNode(), recvNode);
		}

		callParam.setSendRequest(newMessage);

		if ((newMessage.header().commandCode() == 272) && (recvNode.getNodeType() == 1)) {
			String serviceContextId = callParam.getServiceContextId();

			int encrypteCode = BonEncrypt.encrypteMessage(newMessage, serviceContextId, null);
			if (encrypteCode != 0) {
				String errorMsg = encrypteCode + ",encrypt message failed:" + callParam.getSessionId();
				LOG.warn(errorMsg);
			}

			int signCode = BonSignature.signMessage(newMessage, serviceContextId, null);
			if (signCode != 0) {
				String errorMsg = signCode + ",sign message failed:" + callParam.getSessionId();
				LOG.warn(errorMsg);
			}
		}
		try {
			sendRequest(newMessage, toPeer, new Object());
			MessageDetail md = new MessageDetail();
			md.setBusinessCallId(callParam.getBussinessCallId());
			md.setSessionId(callParam.getSessionId());
			md.setServiceContextId(callParam.getServiceContextId());
			md.setMessageType("CCR");
			md.setOpType("send");
			md.setHostId(toPeer.hostID());
			md.setMsgTime(new Date(newMessage._rt0));
			md.setDetails(newMessage.getDictionary().toString());
			MessageDetailCacheManager.getInstance().add(md);
		} catch (StaleConnectionException e) {
			e.printStackTrace();
		} catch (NotARequestException e) {
			e.printStackTrace();
		}

		return 0;
	}

	public void sendAnswer(Peer fromPeer, CallParam callParam) {
		Message request = (Message) callParam.getRequest();
		String ccrOriginalHost = DccAvpHelper.getMessageOriginalHost(request);
		
		
		int s_rc = DccAvpHelper.getCCAMessageServiceResultCode((Message) callParam.getResponse(),callParam.getSessionId());
		callParam.setServiceResultCode(s_rc);
		
		int rc = DccAvpHelper.getCCAMessageResultCode((Message) callParam.getResponse());
		callParam.setRetCode(rc);

		int encrypteCode = BonEncrypt.encrypteMessage((Message) callParam.getResponse(), callParam.getServiceContextId(), ccrOriginalHost);
		if (encrypteCode != 0) {
			String errorMsg = encrypteCode + ",encrypt message failed:" + callParam.getSessionId();
			LOG.warn(errorMsg);
		}
		int signCode = BonSignature.signMessage((Message) callParam.getResponse(), callParam.getServiceContextId(), ccrOriginalHost);
		if (signCode != 0) {
			String errorMsg = signCode + ",sign message failed:" + callParam.getSessionId();
			LOG.warn(errorMsg);
		}
		try {
			answer((Message) callParam.getResponse(), fromPeer);

			MessageDetail md = new MessageDetail();
			md.setBusinessCallId(callParam.getBussinessCallId());
			md.setSessionId(callParam.getSessionId());
			md.setMessageType("CCA");
			md.setServiceContextId(callParam.getServiceContextId());
			md.setOpType("send");
			md.setHostId(fromPeer.hostID());
			md.setMsgTime(new Date(((Message) callParam.getResponse())._rt0));
			md.setDetails(((Message) callParam.getResponse()).getDictionary().toString());
			MessageDetailCacheManager.getInstance().add(md);
		} catch (NotAnAnswerException e) {
			e.printStackTrace();
		}
	}

	public Message directReplyCCA(Message request, Peer peer, int resultCode, AVP failedAVP) {
		Message answer = new Message();
		answer.prepareResponse(request);
		MessageDictionary md = request.getDictionary();
		md.prepareResponseAVP(answer);

		answer.add(new AVP_Unsigned32(268, resultCode));
		if (failedAVP != null) {
			failedAVP.setMandatory(true);
			answer.add(failedAVP);
		}
		addOurHostAndRealm(answer);
		answer.header().setError(true);
		try {
			answer(answer, peer);
		} catch (NotAnAnswerException e) {
			LOG.warn("reply err " + e);
		}
		return answer;
	}
	
	
	public Message directReplyCCA1(Message request, Peer peer, int resultCode, AVP failedAVP) {
		Message answer = new Message();
		answer.prepareResponse(request);
		MessageDictionary md = request.getDictionary();
		md.prepareResponseAVP(answer);

		answer.add(new AVP_Unsigned32(268, 2001));
		answer.add(DccAvpHelper.genDefaultServiceResultAVP(resultCode));
		if (failedAVP != null) {
			failedAVP.setMandatory(true);
			answer.add(failedAVP);
		}
		addOurHostAndRealm(answer);
		answer.header().setError(true);
		try {
			answer(answer, peer);
		} catch (NotAnAnswerException e) {
			LOG.warn("reply err " + e);
		}
		return answer;
	}

	public Message directReplyCCAWithServiceResultCode(Message request, Peer peer, int serviceResultCode, AVP failedAVP) {
		Message answer = new Message();
		answer.prepareResponse(request);
		MessageDictionary md = request.getDictionary();
		md.prepareResponseAVP(answer);
		answer.add(new AVP_Unsigned32(268, 2001));
		answer.add(DccAvpHelper.genDefaultServiceResultAVP(serviceResultCode));
		if (failedAVP != null) {
			failedAVP.setMandatory(true);
			answer.add(failedAVP);
		}
		addOurHostAndRealm(answer);
		answer.header().setError(true);
		try {
			answer(answer, peer);
		} catch (NotAnAnswerException e) {
			LOG.warn("reply err " + e);
		}
		return answer;
	}

	protected Message genBonCallCCR(CallParam callParam) {
		ProtocalConvertResult<AVP> result = null;

		XmlMessage req = (XmlMessage) callParam.getRequest();
		try {
			Document document = Comutils.parseText(req.getDATA());
			result = DataExchangerImpl.getInstance().converToCCRRequest(callParam.getServiceContextId(), document.getRootElement());
		} catch (Exception e) {
			e.printStackTrace();
			callParam.setRetCode(11300);
			callParam.setResultMsg("convert err :" + e.toString());
			callParam.getAsynHandler().onCallResult(callParam);
			return null;
		}
		if (result.getRetCode() != 0) {
			InprogressMessageQueue.getInstance().removeProgress(callParam.getMessageUniqueID());
			callParam.setRetCode(result.getRetCode());
			callParam.setResultMsg(result.getErrMsg());
			callParam.getAsynHandler().onCallResult(callParam);
			return null;
		}

		Message msg = new Message();
		DccAvpHelper.creatComMessageBeforCCR(msg, callParam.getSessionId(), callParam.getServiceContextId());

		AVP oH = new AVP_OctetString(264, callParam.getUserName());
		oH.setMandatory(true);
		msg.add(oH);

		AVP oR = new AVP_OctetString(296, DccAvpHelper.getRealmPart(callParam.getUserName()));
		oR.setMandatory(true);
		msg.add(oR);

		DccAvpHelper.addBusinessAVPs((AVP) result.getResult(), msg);
		return msg;
	}

	public SGWChannel getSGWChannel() {
		return this.sgwChannel;
	}

	public boolean isFluxContrlToReject(String hostId) {
		if (!FluxControlService.getInstance().checkAllowedConnect(hostId)) {
			LOG.warn("11033|" + hostId + " has been dennied,please wait some more minutes ");
			return true;
		}
		return false;
	}

	public boolean isCongestionToReject() {
		if (SpringContext.getInstance().getCongestionBean().getCurCrawedLevel() == 9999) {
			LOG.warn("11007|system congestioned to DOS(denial of service) level");
			return true;
		}
		return false;
	}

	public void addPersist(Node node) {
		LOG.info("add persist peer " + node.getHostId());
		try {
			if ((node.getNodeType() == 1) || (node.getNodeType() == 2) || (node.getNodeType() == 3)) {
				Peer peer = getPeerFormNode(node);
				if (peer != null) {
					LOG.info("connect to persist peer " + node.getHostId() + "[" + node.getUri() + "]");
					addPersistentPeer(peer);
				}
			}
		} catch (Exception e) {
			LOG.error("connect persist peer err " + e);
		}
		this.persistNodes.put(node.getHostId(), node);
	}

	public void deletePersist(Node node) {
		LOG.info("remove persist node peer " + node.getHostId() + "[" + node.getUri() + "]");
		Peer peer = getNodeConnect(node.getHostId());
		if (peer != null)
			try {
				closePeer(peer);
			} catch (Exception e) {
				LOG.warn("remove persist Node err " + e.toString());
			}
		else {
			peer = getPeerFormNode(node);
		}
		removePersistentPeer(peer);
		this.persistNodes.remove(node.getHostId());
	}

	public void deleteAllowedNode(Node node) {
		LOG.info("remove allowed node peer " + node.getHostId() + "[" + node.getUri() + "]");
		Peer peer = getNodeConnect(node.getHostId());
		if (peer != null)
			try {
				closePeer(peer);
			} catch (Exception e) {
				LOG.warn("remove allowed Node err " + e.toString());
			}
		else {
			peer = getPeerFormNode(node);
		}
		removePersistentPeer(peer);
		this.allowedNodes.remove(node.getHostId());
	}

	public void logHandleReceiveMessage(Message msg, Connection conn) {
		MessageDictionary md = msg.getDictionary();

		if ((LOG.isInfoEnabled()) && (msg.header().commandCode() != 280))
			LOG.info("Received message from " + conn.toString() + ",Plain Text:\n" + md.toString());
	}

	public void logReceivingMessage(Message msg, Connection conn) {
		if (LOG.isInfoEnabled())
			LOG.info("Receiving message from " + conn.toString() + "," + msg.toString());
	}

	public void logPrepareSendMessage(Message msg, Connection conn) {
		MessageDictionary messageDictionary = msg.getDictionary();
		if ((LOG.isInfoEnabled()) && (msg.header().commandCode() != 280))
			LOG.info("Sending message to " + conn.toString() + ",Plain Text:\n" + messageDictionary.toString());
	}

	public void logAfterSendMessage(Message msg, Connection conn) {
		if (LOG.isInfoEnabled())
			LOG.info("Send message to " + conn._peer.hostID() + ", " + msg.toString());
	}

	public void updateAndInfoNodeStatus(Node node, boolean isConnected) {
		node.setNewStatus(isConnected ? 1 : 0, System.currentTimeMillis());
		if (!isConnected) {
			LOG.warn("linkexception|17001|" + node.getNodeCode() + "|" + node.getHostId() + " connection closed");
		}

		StringBuilder sb = new StringBuilder();
		sb.append("update Node set status =").append(node.getStatus());
		sb.append(",statusTime = sysDate");
		sb.append(" where id= ").append(node.getId());
		SpringContext.getInstance().getDataStoreService().addHSQL(sb.toString(), true);
	}

	public Peer getNodeConnect(String nodeName) {
		Peer toPeer = null;
		for (Peer p : getActivePeers()) {
			if (p.hostID().equals(nodeName)) {
				toPeer = p;
			}
		}

		return toPeer;
	}

	private Peer getPeerFormNode(Node node) {
		Peer peer = null;
		try {
			URI uri = new URI(node.getFirstUri());
			String hostID = node.getHostId();
			String realm = node.getRealm();
			peer = PeerFactory.getInstance().getPeerServer(hostID, realm, uri);
		} catch (URISyntaxException e) {
			e.printStackTrace();
		} catch (UnsupportedURIException e) {
			e.printStackTrace();
		}
		return peer;
	}

	public int getNodeSendQueueSize(String nodeHostId) {
		Peer peer = getNodeConnect(nodeHostId);
		if (peer != null) {
			return super.getSenderQueueSize(peer);
		}
		return 0;
	}

	private void initParams() {
		this.monitorConfKeys.clear();
		this.monitorConfKeys.add("Diameter.CCA.Timeout");
		this.monitorConfKeys.add("Common.Nodes");
		this.monitorConfKeys.add("Diameter.PrintMessageDetail");
		this.monitorConfKeys.add("agreement.dcc.timeout_linkwait");
		this.monitorConfKeys.add("agreement.dcc.timeout_dwr");
		this.monitorConfKeys.add("agreement.dcc.outcount_dwr");
		this.monitorConfKeys.add("agreement.dcc.timeout_answer");
		this.monitorConfKeys.add("agreement.dcc.repetition_times");
		this.monitorConfKeys.add(SysParamBean.KEY_AGREEMENT_DCC_CheckStrick);
	}

	public void onConfigUpdated(String key) {
		if (key.startsWith("Common.Nodes"))
			loadPeerNodes();
		else if ((key.startsWith("agreement.dcc.")) || (key.startsWith("Diameter.CCA.Timeout")))
			loadConfParams();
		else if (key.startsWith("Diameter.CCA.Timeout"))
			loadConfParams();
	}

	private void loadPeerNodes() {
		ConcurrentHashMap<String, Node> map = ConfigService.name2NodeMap;
		if (this.persistNodes != null) {
			Map<String, Node> allPersistNodes = new HashMap<String, Node>();
			Map<String, Node> allAllowedNodes = new HashMap<String, Node>();

			for (Node node : map.values()) {
				if (node.getChannelId() == this.sgwChannel.getId()) {
					if (node.getConnectType() == 1)
						allPersistNodes.put(node.getHostId(), node);
					else if (node.getConnectType() == 0) {
						allAllowedNodes.put(node.getHostId(), node);
					}
				}
			}

			for (Node node : allPersistNodes.values()) {
				if (!this.persistNodes.containsKey(node.getHostId())) {
					addPersist(node);
				}
			}

			List<Node> newAllowedNodes = new LinkedList<Node>();
			for (Node node : allAllowedNodes.values()) {
				if (!this.allowedNodes.containsKey(node.getHostId())) {
					newAllowedNodes.add(node);
				}
			}

			for (Node node : this.persistNodes.values()) {
				if (!allPersistNodes.containsKey(node.getHostId())) {
					deletePersist(node);
				}
			}

			for (Node node : this.allowedNodes.values())
				if (!allAllowedNodes.containsKey(node.getHostId())) {
					deleteAllowedNode(node);
				}
		}
	}

	private void loadConfParams() {
		this.dwrTimeOut = this.confSrv.getIntValue("agreement.dcc.timeout_dwr", 60);
		this.dwrTimeOutCount = this.confSrv.getIntValue("agreement.dcc.outcount_dwr", 3);
		this.reConnectTime = this.confSrv.getIntValue("agreement.dcc.timeout_linkwait", 10);

		this.ccrSendTimeout = this.confSrv.getIntValue("agreement.dcc.timeout_answer", 60);
		this.ccrRepeatCount = this.confSrv.getIntValue("agreement.dcc.repetition_times", 0);

		this.checkStrickCheckCCR = this.confSrv.getBooleanValue(SysParamBean.KEY_AGREEMENT_DCC_CheckStrick, false);

		setCheckMessageFormat(this.checkStrickCheckCCR);
		setConnectionTimeout(this.dwrTimeOutCount, this.dwrTimeOut, 0L);
		setReconnectRetryTime(this.reConnectTime);
		setRetransmitParameter(this.ccrRepeatCount, this.ccrSendTimeout * 1000);

		LOG.info(this.sgwChannel.getChannelHost() + " connection DWR timeout: " + this.dwrTimeOut + "s" + ", max DWR timeout count: " + this.dwrTimeOutCount + ", reconnect time: " + this.reConnectTime + "s" + ", ccr send timeout " + this.ccrSendTimeout + ", ccr repeat count:" + this.ccrRepeatCount);
	}

	public Set<String> monitorConfKeys() {
		return this.monitorConfKeys;
	}

	public static NodeSettings buildDCCNodeParams(SGWChannel dccChannel, ConfigService confSrv) {
		URI addr;
		try {
			addr = new URI(dccChannel.getUri());
		} catch (URISyntaxException e) {
			e.printStackTrace();
			return null;
		}
		String ipAddr = addr.getHost();
		if ((ipAddr.equals("127.0.0.1")) || (ipAddr.equalsIgnoreCase("localhost"))) {
			ipAddr = "0.0.0.0";
		}
		int port = addr.getPort();
		Capability capability = new Capability();
		String[] items = confSrv.getValue("Diameter.Capability.AuthApps", "4").split(",");
		for (String it : items) {
			int id = Integer.parseInt(it);
			capability.addAuthApp(id);
		}
		items = confSrv.getValue("Diameter.Capability.AcctApps", "4").split(",");
		for (String it : items) {
			int id = Integer.parseInt(it);
			capability.addAcctApp(id);
		}
		items = confSrv.getValue("Diameter.Capability.SupportedVendors", "11,193,10415,81000").split(",");
		for (String it : items) {
			int id = Integer.parseInt(it);
			capability.addSupportedVendor(id);
		}
		String params = confSrv.getValue("Diameter.Capability.SupportedVendorsAuthApps", null);
		if ((params != null) && (!params.trim().equals(""))) {
			items = params.split(";");
			for (String it : items) {
				String[] pair = it.split(",");
				int id = Integer.parseInt(pair[0]);
				int app = Integer.parseInt(pair[1]);
				capability.addVendorAuthApp(id, app);
			}
		}
		params = confSrv.getValue("Diameter.Capability.SupportedVendorsAcctApps", null);
		if ((params != null) && (!params.trim().equals(""))) {
			items = params.split(";");
			for (String it : items) {
				String[] pair = it.split(",");
				int id = Integer.parseInt(pair[0]);
				int app = Integer.parseInt(pair[1]);
				capability.addVendorAcctApp(id, app);
			}
		}
		int vendorID = confSrv.getIntValue("Diameter.VendorID", 11);
		try {
			return new NodeSettings(dccChannel.getChannelHost(), dccChannel.getRealm(), vendorID, capability, ipAddr, port, "BON SGW System Supported By HP", 16777216);
		} catch (InvalidSettingException e) {
			e.printStackTrace();
		}
		return null;
	}

	public int getThreadPoolActiveCount() {
		return 0;
	}
}